package cn.hinus.iomodel;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created by hinus on 2021/9/7.
 */
public class MTReactor implements Runnable {
    public static void main(String[] args) throws IOException {
        new Reactor(54321).run();
    }

    final ServerSocketChannel serverChannel;
    final Selector selector;

    final ExecutorService executorService;

    public MTReactor(int port) throws IOException {
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress("127.0.0.1", port));

        selector = Selector.open();
        SelectionKey key = serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        key.attach(new MTAcceptor());

        executorService = new ThreadPoolExecutor(4, 4, 0, TimeUnit.MILLISECONDS,
                new LinkedBlockingDeque<>());
    }

    public void dispatch(SelectionKey key) {
        Runnable r = (Runnable)key.attachment();
        executorService.execute(r);
    }

    @Override
    public void run() {
        try {
            while (true) {
                selector.select();
                Iterator ite = selector.selectedKeys().iterator();

                while (ite.hasNext()) {
                    System.out.println("ready to do dispatch");
                    dispatch((SelectionKey)ite.next());
                    ite.remove();
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    class MTAcceptor implements Runnable {
        @Override
        public synchronized void run() {
            try {
                SocketChannel sock = serverChannel.accept();
                if (sock != null)
                    new MTHandler(selector, sock);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}

final class MTHandler implements Runnable {
    final SocketChannel socket;
    final SelectionKey key;
    ByteBuffer input = ByteBuffer.allocate(128);
    ByteBuffer output = ByteBuffer.allocate(128);

    final int READING = 0;
    final int SENDING = 1;
    final int PROCESSING = 2;

    volatile int state = READING;


    public MTHandler(Selector selector, SocketChannel socket) throws IOException {
        this.socket = socket;
        this.socket.configureBlocking(false);
        key = this.socket.register(selector, 0);
        key.attach(this);
        key.interestOps(SelectionKey.OP_READ);
        selector.wakeup();
    }

    @Override
    public void run() {
        try {
            if (state == READING) {
                read();
            }
            else if (state == SENDING) {
                write();
            }
        } catch (IOException e) {
            try {
                socket.close();
            } catch (IOException e1) {
                e1.printStackTrace();
            }
            key.cancel();
        }
    }

    public synchronized void read() throws IOException {
        if (state != READING)
            return;

        state = PROCESSING;
        input.clear();

        // read may fail even SelectionKey is readable
        int n = socket.read(input);
        while (n == 0) {
            n = socket.read(input);
        }

        if (n == -1) {
            socket.close();
            key.cancel();
            return;
        }

        System.out.println("received " + n + " bytes from client");
        input.flip();
        byte[] buf = new byte[32];
        input.get(buf, 0, input.remaining());
        System.out.println("received msg is " + new String(buf));

        state = SENDING;
        key.interestOps(SelectionKey.OP_WRITE);
    }

    public synchronized void write() throws IOException {
        if (state != SENDING)
            return;

        state = PROCESSING;

        output.clear();
        output.put("nihao".getBytes());

        output.flip();
        socket.write(output);
        state = READING;
        key.interestOps(SelectionKey.OP_READ);
    }
}

